package com.cmsz.collection.service;

import com.cmsz.collection.util.DateUtil;
import net.sf.json.JSONObject;
import org.apache.kafka.clients.producer.*;
import org.junit.Before;
import org.junit.Test;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Random;

/**
 * @author liupeng
 * @date 2020/1/15
 * @description
 */
public class CreateTestData {
    private Producer kafkaProducer;

    private Map<String,Object> map = new HashMap<>();
    private Map<String,String> metadata = new HashMap<>();
    private Map<String,String> agent = new HashMap<>();
    private Map<String,String> log = new HashMap<>();
    private Map<String,String> fields = new HashMap<>();

    private static final String[] hostnames = {"192.168.121.193","192.168.121.194","192.168.121.195"};
    private static final String[] applicationNames = {"oc-crm-front-dcc","oc-crm-front-http","oc-core"};
    private static final String[] log_topics = {"info","alarm","message"};
    private static final String info = "INFO#2019-11-25 19:11:54.770#oc-front-server_192.168.158.238_8081#Serial:01302220191125191146112648637524#ActivityCode:013022#ReqSys:0068#Province:200#com.cmsz.oc.front.utils.HttpClient:http协议异常|http协议异常";
    private static final String alarm = "BL##WARNING#20191202122325#172.17.42.1#UPAY#TXN#merchant-front#172.17.42.1_9110#ReqSys:9999#com.cmit.merchant.service.OrderNotifyController##||OrderNo:201912027708583093429311|PayTrans:99992019120211163876040236686887##发送结果通知异常,并已到达最大重发次数##LB";
    private static final String message = "2019-12-02 12:21:19.618#merchant-front_172.17.42.1_9110#Send#ON_REQ|发送订单结果通知报文至商户: PayTrans=99992019120211163876040236686887|20191202139571924169899146755357|request_params={\"ReqSys\": \"0001\",\"OrderDate\": \"20191202\",\"ReqDateTime\": \"20191202122119\",\"OrderNo\": \"201912027708583093429311\",\"Payment\": \"1\",\"PaymentType\": \"ALIPAY-WEB\",\"PayTrans\": \"99992019120211163876040236686887\",\"CustomParam\": \"key=valuue|key2=value2\",\"SettleDate\": \"20191202\",\"OrderStatusCode\": \"015A16\",\"OrderStatusDesc\": \"用户超时未支付\"}&sign={\"CerID\":\"56445F07\",\"SignValue\":\"lJAbWKzG0vM4nAvBjojVH+s+k0Tze79xeAVbqieZYuuNSNXvwmiUiM9kfqm+yPHH2Ijr1K2h832tJXkbasQYL7fx/PSDrN2HZb5leOlKSZxHGdXtSEyq8r5jvbR+qv8tYHWaH6RZXuWbs7La1QLzeM12jKJZosReqyJE/h/8J5I=\"}";


    @Before
    public void setUp() throws Exception {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "192.168.121.195:9092,192.168.121.196:9092,192.168.121.197:9092");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        kafkaProducer = new KafkaProducer(properties);
        initMap();
    }

    @Test
    public void kafkaProducer() throws Exception {
        Random random = new Random();
        for (int i = 0; i < 100000; i++) {
            log.put("offset",i+"");
            fields.put("log_topic",log_topics[random.nextInt(3)]);
            kafkaProducer.send(get(fields.get("log_topic"),random), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        exception.printStackTrace();
                    }
                }
            });
        }
    }



    private ProducerRecord get(String topic, Random random ){
        map.put("@timestamp", DateUtil.getUTCTime());
        fields.put("hostname",hostnames[random.nextInt(3)]);
        fields.put("applicationName",applicationNames[random.nextInt(3)]);

        switch (topic){
            case "info":
                map.put("message",info);
                break;
            case "alarm":
                map.put("message",alarm);
                break;
            case "message":
                map.put("message",message);
                break;
            default:
                map.put("message","");
        }
        JSONObject jsonObject = JSONObject.fromObject(map);
        return new ProducerRecord<String,String>(topic,jsonObject.toString());
    }

    private void initMap(){
        map.put("ecs","\"version\":\"1.1.0\"");
        map.put("host","\"name\":\"NameNode\"");
        map.put("input","\"type\":\"log\"");
        map.put("@metadata",metadata);
        map.put("agent",agent);
        map.put("log",log);
        map.put("fields",fields);
        metadata.put("beat","filebeat");
        metadata.put("type","log");
        metadata.put("version","7.5.1");
        agent.put("ephemeral_id","f489fb58-76fb-4932-9b10-a8876a5c23c9");
        agent.put("hostname","NameNode");
        agent.put("id","6cc0c0e5-367e-461f-b067-22c5def87f1f");
        agent.put("version","7.5.1");
        agent.put("type","filebeat");
        log.put("offset","88501");
        log.put("file","\"path\":\"/app/logs/oc-crm-front-dcc/20200106/info/info.log.2019-11-25.0\"");
    }
}
